CloudFlow集成OTS实现流程中各种数据的读写和持久化存储

云工作流可以在流程中集成表格存储（Tablestore），通过集成Tablestore，您可以将流程运行中的各种数据持久化到Tablestore的某个表中，也可以将某个表的数据作为流程的数据源。本文分别从数据读取和数据写入两种场景，介绍云工作流集成Tablestore的示例代码、参数说明和集成输出。

背景信息

Tablestore面向海量结构化数据提供Serverless表存储服务，同时针对物联网场景深度优化提供一站式的IoTstore解决方案。适用于海量账单、IM消息、物联网、车联网、风控、推荐等场景中的结构化数据存储，提供海量数据低成本存储、毫秒级的在线数据查询和检索以及灵活的数据分析能力。更多信息，请参见什么是表格存储。

执行模式

云工作流集成Tablestore时，仅支持请求响应（RequestComplete）集成模式。

数据读取

GetRow

根据指定的主键读取单行数据。

集成示例

...
  - Type: Task
    Name: ots_get
    Action: Tablestore:GetRow
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      primaryKey.$: $Input.primaryKey
      columnsToGet:
        - COL1

参数说明

参数	必填	说明	示例
region	否	Tablestore实例所在地域，默认为当前流程所在的地域。	cn-hangzhou
instanceName	是	Tablestore实例名称。	test-instance-name
tableName	是	Tablestore中的表名。	test_table_name
primaryKey	是	数据主键。由于Tablestore支持多个主键列，这里使用一个对象的形式给出各个主键列名和对应的值。	`primaryKey: pkCol1: val1 pkCol2: val2`
columnsToGet	否	需要查询和返回的列名，通过一个数组指定每个需要返回的列名。	`columnsToGet: - col1 - col2`
startColumn	否	指定读取的起始列，主要用于宽行读。列的顺序按照列名的字典序排序。返回的结果中包含当前起始列。	col1
endColumn	否	读取的结束列，返回的结果中不包含当前结束列。列的顺序按照列名的字典序排序。	col10

集成输出

{
    "requestId":"00060e6a-7bdb-786f-5d68-******",
    "row":{
        "COL1":"val2",
        "PK1":"aaa",
        "PK2":1699949934946000
    }
}

GetRange

读取指定主键范围内的数据。

集成示例

...
  - Type: Task
    Name: ots_get
    Action: Tablestore:GetRange
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      startPrimaryKey:
        - key: pkCol1
          valueType: value
          value.$: $Input.value1
        - key: pkCol2
          valueType: min
      endPrimaryKey:
      	- key: pkCol1
        	valueType: value
        	value.$: $Input.value1
        - key: pkCol2
      		valueType: max
      columnsToGet:
        - COL1

参数说明

参数	必填	说明	示例
region	否	Tablestore实例所在地域，默认为当前流程所在的地域。	cn-hangzhou
instanceName	是	Tablestore实例名称。	test-instance-name
tableName	是	Tablestore的表名。	test_table_name
startPrimaryKey	是	本次范围读取的起始主键。需要一个对象对不同主键列做进一步说明： valueType：主键值类型。min=最小值，max=最大值，value=指定值，auto=自增。 value：当`valueType = value`时，通过此字段指定具体值。	`startPrimaryKey: - key: pkCol1 valueType: value value: val1 - key: pkCol2 valueType: min`
endPrimaryKey	是	本次范围读取的结束主键。	`endPrimaryKey: - key: pkCol1 valueType: value value: val1 - key: pkCol2 valueType: max`
columnsToGet	否	需要查询和返回的列名，通过一个数组指定每个需要返回的列名。	`columnsToGet: - col1 - col2`
startColumn	否	指定读取数据的起始列，主要用于宽行读。列的顺序按照列名的字典序排序。返回的结果中包含当前起始列。	col1
endColumn	否	读取数据的结束列，返回的结果中不包含当前结束列。列的顺序按照列名的字典序排序。	col10
limit	否	本次读取最多返回的行数。取值必须大于0。	10

集成输出

{
    "requestId":"00060e6a-88c3-ecd1-c1cb-******",
    "rows":[
        {
            "COL1":"val2",
            "PK1":"aaa",
            "PK2":1699881177699000
        },
        {
            "COL1":"val2",
            "PK1":"aaa",
            "PK2":1699881296484000
        }
    ]
}

数据写入

PutRow

新写入一行数据。如果该行已存在，则先删除原行数据（原行的所有列以及所有版本的数据），再写入新行数据。

集成示例

...
  - Type: Task
    Name: ots
    Action: Tablestore:PutRow
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      primaryKey:
        PK1: aaa
      rowData:
        COL1: val1
      condition: IGNORE

参数说明

参数	必填	说明	示例
region	否	Tablestore实例所在地域，默认为当前流程所在的地域。	cn-hangzhou
instanceName	是	Tablestore实例名称。	test-instance-name
tableName	是	Tablestore的表名。	test_table_name
primaryKey	是	数据主键。由于Tablestore支持多个主键列，这里使用一个对象的形式给出各个主键列名和对应的值。	`primaryKey: pkCol1: val1 pkCol2: val2`
rowData	否	写入的行数据，不需要包括主键。	`rowData: col3: val3 col4: val4`
condition	否	在数据写入前是否进行存在性检查。取值范围如下： IGNORE（默认）：不做行存在性检查。 EXPECT_EXIST：期望行存在。 EXPECT_NOT_EXIST：期望行不存在。	IGNORE

集成输出

{
    "requestId":"00060e6a-8c81-9ac4-5d68-******",
    "row":{
        "COL1":"val2",
        "PK1":"aaa",
        "PK2":1704700646764000
    }
}

UpdateRow

更新一行数据，可以增加和删除一行中的属性列，删除属性列指定版本的数据，或者更新已存在的属性列的值。如果更新的行不存在，则新增一行数据。

集成示例

...
  - Type: Task
    Name: ots_update
    Action: Tablestore:UpdateRow
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      primaryKey.$: $Input.primaryKey
      rowData:
        COL1: val2
      condition: EXPECT_EXIST

参数说明

参数	必填	说明	示例
region	否	Tablestore实例所在地域，默认为当前流程所在的地域。	cn-hangzhou
instanceName	是	Tablestore实例名称。	test-instance-name
tableName	是	Tablestore的表名。	test_table_name
primaryKey	是	数据主键。由于Tablestore支持多个主键列，这里使用一个对象的形式给出各个主键列名和对应的值。	`primaryKey: pkCol1: val1 pkCol2: val2`
rowData	否	写入的行数据，不需要包括主键。	`rowData: col3: val3 col4: val4`
condition	否	在数据写入前是否进行行存在性检查。取值范围如下： IGNORE（默认）：不做行存在性检查。 EXPECT_EXIST：期望行存在。 EXPECT_NOT_EXIST：期望行不存在。	IGNORE

集成输出

{
    "requestId":"00060e6a-8c81-9ac4-5d68-******",
    "row":{
        "COL1":"val2",
        "PK1":"aaa",
        "PK2":1704700646764000
    }
}

DeleteRow

删除一行数据。

集成示例

...
  - Type: Task
    Name: ots_get
    Action: Tablestore:DeleteRow
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      primaryKey.$: $Input.primaryKey

参数说明

参数	必填	说明	示例
region	否	Tablestore实例所在地域，默认为当前流程所在的地域。	cn-hangzhou
instanceName	是	Tablestore实例名称。	test-instance-name
tableName	是	Tablestore的表名。	test_table_name
primaryKey	是	数据主键。由于Tablestore支持多个主键列，这里使用一个对象的形式给出各个主键列名和对应的值。	`primaryKey: pkCol1: val1 pkCol2: val2`
condition	否	在数据写入前是否进行行存在性检查。取值范围如下： IGNORE（默认）：不做行存在性检查。 EXPECT_EXIST：期望行存在。 EXPECT_NOT_EXIST：期望行不存在。	IGNORE

集成输出

{
    "requestId":"00060e6a-939b-965b-1069-******"
}